package com.itheima.kafka.example2;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafkaStreams;

import java.time.Duration;
import java.util.Arrays;

/**
 * 示例2：Kafka Streams 单词计数处理器
 * 
 * 学习目标：
 * 1. 理解 KStream 的使用
 * 2. 学习流式数据转换（flatMapValues, groupBy, count）
 * 3. 理解时间窗口的概念
 */
@Slf4j
@Configuration
@EnableKafkaStreams
public class StreamWordCountProcessor {

    /**
     * 创建单词计数的 Stream 处理流程
     * 
     * 处理流程：
     * 1. 从 word-count-input topic 读取文本
     * 2. 将文本按空格分割成单词
     * 3. 对单词进行分组
     * 4. 在时间窗口内统计每个单词的出现次数
     * 5. 输出结果到 word-count-output topic
     */
    @Bean
    public KStream<String, String> wordCountStream(StreamsBuilder builder) {
        
        // 1. 从输入 topic 读取数据流
        KStream<String, String> textStream = builder.stream(
            "word-count-input",
            Consumed.with(Serdes.String(), Serdes.String())
        );
        
        // 2. 处理流程
        KStream<String, String> wordCountResult = textStream
            // 2.1 将每行文本分割成单词
            .flatMapValues(value -> {
                log.info("🔄 [Stream] 处理文本: {}", value);
                return Arrays.asList(value.toLowerCase().split("\\s+"));
            })
            // 2.2 按单词分组
            .groupBy(
                (key, word) -> word,
                Grouped.with(Serdes.String(), Serdes.String())
            )
            // 2.3 定义时间窗口（10秒窗口）
            .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
            // 2.4 统计每个单词的出现次数
            .count()
            // 2.5 转换回 KStream
            .toStream()
            // 2.6 格式化输出
            .map((windowedKey, count) -> {
                String word = windowedKey.key();
                String windowInfo = String.format("[窗口: %s - %s]",
                    windowedKey.window().startTime(),
                    windowedKey.window().endTime());
                
                String result = String.format("单词: '%s' | 出现次数: %d | %s", 
                    word, count, windowInfo);
                
                log.info("📊 [Stream结果] {}", result);
                
                return KeyValue.pair(word, result);
            });
        
        // 3. 将结果输出到目标 topic
        wordCountResult.to(
            "word-count-output",
            Produced.with(Serdes.String(), Serdes.String())
        );
        
        log.info("✅ [Stream] 单词计数Stream已启动");
        
        return wordCountResult;
    }
}

